package com.wuwangfu.state;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: jcshen
 * @Date: 2023-03-03
 * @PackageName: com.wuwangfu.state
 * @ClassName: MapStated
 * @Description:
 * @Version: 1.0.0
 *
 * keyBy之后，用来存储K-V类型的状态，叫做KeyState
 * KeyState中类：
 *  ValueState<T>：value是一个基本类型、集合类型、自定义类型
 *  MapState<小K,V>：存储的是KV类型，大K->(小K,小V)
 *  ListState：Value是一个list集合
 *
 * MapState源码
 * org.apache.flink.runtime.state.heap.HeapMapState#get(java.lang.Object)
 *      org.apache.flink.runtime.state.heap.StateTable#get(java.lang.Object)
 *          org.apache.flink.runtime.state.heap.StateTable#get(java.lang.Object, int, java.lang.Object)
 *              org.apache.flink.runtime.state.heap.CopyOnWriteStateMap#get(java.lang.Object, java.lang.Object)
 */
public class MapStated {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000));
        /**
         * 辽宁省,沈阳市,1000
         * 辽宁省,铁岭市,2000
         * 辽宁省,沈阳市,2000
         * 河北省,廊坊市,2000
         * 河北省,保定市,2000
         * 河北省,保定市,2000
         */
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Tuple3<String, String, Double>> tpStream = line.map(new MapFunction<String, Tuple3<String, String, Double>>() {
            @Override
            public Tuple3<String, String, Double> map(String value) throws Exception {
                String[] fields = value.split(",");

                return Tuple3.of(fields[0], fields[1], Double.parseDouble(fields[2]));
            }
        });
        //分组
        KeyedStream<Tuple3<String, String, Double>, String> keyed = tpStream.keyBy(t -> t.f0);
        //
        SingleOutputStreamOperator<Tuple3<String, String, Double>> result = keyed.process(new KeyedProcessFunction<String, Tuple3<String, String, Double>, Tuple3<String, String, Double>>() {
            //在JobManager中赋值后，会在反序列化后有值，希望在open方法后，从状态后端直接取值
            //transient 不希望参与序列化和反序列化
            private transient MapState<String, Double> mapstate;

            @Override
            public void open(Configuration parameters) throws Exception {

                //定义一个状态描述器
                MapStateDescriptor<String, Double> mapDesc = new MapStateDescriptor<>("kv-state", String.class, Double.class);
                //初始化或恢复历史状态
                mapstate = getRuntimeContext().getMapState(mapDesc);
            }

            @Override
            public void processElement(Tuple3<String, String, Double> value, Context context, Collector<Tuple3<String, String, Double>> output) throws Exception {
                String city = value.f1;
                Double money = value.f2;
                //状态值
                Double histMoney = mapstate.get(city);
                if (histMoney == null) {
                    histMoney = 0.0;
                }
                //累计
                Double total = money + histMoney;
                //更新状态
                mapstate.put(city, total);
                //赋值
                value.f2 = total;
                //返回
                output.collect(value);

            }
        });

        result.print();

        env.execute();
    }
}
